package edu.hfuu.voicedemo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.FileOutputStream;
import java.io.IOException;
import java.util.List;

public class QuerySound {

    //表名 soudata_timeorder (包含声音）
    private static final byte[] TABLE_soudata_timeorder = Bytes.toBytes("testNamespace:soudata_timeorder");
    private static final byte[] TABLE_channel_index_soudata = Bytes.toBytes("testNamespace:channel_index_soudata");

    public static void main(String[] args) throws IOException {
        QuerySound querySound = new QuerySound();

   //     querySound.getData();    //查询第2张表，按指定rowkey 查询所有列

//        querySound.getSouData();  //询第1张表，按指定rowkey 查询指定列族的指定列——根据时间获取声音数据
//
//        querySound.scanRange2();  //扫描第2张表（可指定rowkey范围）
//
        querySound.scanRange1("23_0_0_1635070172");   //扫描第1张表，批量获取（某一段时间）声音文件

    }

    /**
     * 扫描表1，批量获取声音文件
     */
    public void scanRange1(String rowKey) throws IOException {
        Connection connection = getConnection();
        Table table = connection.getTable(TableName.valueOf("origin_sound_DB:soudata_timeorder"));

        Scan scan = new Scan();
        //添加即可扫描部分获取  【start, stop)
        scan.setStartRow(rowKey.getBytes());
        scan.setStopRow(rowKey.getBytes());

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            String rowkey = Bytes.toString(result.getRow());
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取所有列族名 字节数组
                String familyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());

                //获取列的名称
                String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());

                //过滤其他数据，只取声音数据
                if(familyName.equals("soudata") && columnName.equals("soudData")){
                	byte[] data = cell.getValue();  //cell.getValueArray();
                	System.out.println("数据字节数："+data.length);
                    FileOutputStream fileOutputStream = new FileOutputStream("C:\\audio\\" + rowkey + ".wav");
                    fileOutputStream.write(data);
                    fileOutputStream.close();
                }
            }
        }
    }

    /**
     * 扫描表2 [ startRowKey  endRowKey ) 所有数据
     * @throws IOException
     */
    public void scanRange2() throws IOException {
        Connection connection = getConnection();

        Table table = connection.getTable(TableName.valueOf("testNamespace:channel_index_soudata"));

        Scan scan = new Scan();
        //设置起始和结束的rowkey，即可实现部分扫描 【start, stop)
//        scan.setStartRow("1_1632707457825".getBytes());
//        scan.setStopRow("1_1632910277800".getBytes());

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取所有列族名 字节数组
                String familyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());

                //获取列的名称
                String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());

                if(familyName.equals("collector_info") && columnName.equals("adapter_id") || columnName.equals("channel_id") || columnName.equals("collector_id")){
                    //存的是整形，取出来后也要转为整形
                    int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列族名： " + familyName + ", 列名： " + columnName + ", 列值： " + value);
                } else {
                    //存的是byte 其他两个byte属性：通道ID  采集者
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列族名： " + familyName + ", 列名： " + columnName + ", 列值： " + value);
                }

            }
        }
        table.close();
    }


    /**
     * 查询第一张表，按指定rowkey 查询指定列族的指定列——仅查询声音数据
     */
    public void getSouData() throws IOException {
        Connection connection = getConnection();

        Table table = connection.getTable(TableName.valueOf("testNamespace:soudata_timeorder"));

        Get get = new Get("1632707460625_1".getBytes());
        get.addColumn("soudata".getBytes(), "soudData".getBytes());
        //result对象封装了所有结构数据
        Result result = table.get(get);

        //获取这条数据所有的cell值，实际上这里只有一个cell
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            //获取所有列族名 字节数组
            String familyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());

            //获取列的名称
            String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());

            String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            byte[] bytes = Bytes.toBytes(value);
            FileOutputStream fileOutputStream = new FileOutputStream("C:\\Users\\DouJin\\IdeaProjects\\kafka\\hbasedemo\\data\\out"  + ".wav");
            fileOutputStream.write(bytes);
            fileOutputStream.close();
        }
        table.close();
    }


    /**
     * 查询第二张表，按指定rowkey 查询所有列
     * @throws IOException
     */
    public void getData() throws IOException {

        Connection connection = getConnection();

        Table table = connection.getTable(TableName.valueOf("testNamespace:channel_index_soudata"));

        Get get = new Get(Bytes.toBytes("1_1632707469858"));

        //result对象封装了 该rowkey下的 所有结构数据
        Result result = table.get(get);
//        System.out.println("Get: " + result);

        //获取 1_1632678787566 这条数据所有的cell值
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {

            //获取所有列族名 字节数组
            String familyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());

            //获取列的名称
            String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());

            if(familyName.equals("collector_info") && columnName.equals("adapter_id") || columnName.equals("channel_id") || columnName.equals("collector_id")){
                //存的是整形，取出来后也要转为整形
                int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列族名： " + familyName + ", 列名： " + columnName + ", 列值： " + value);
            } else {
                //存的是byte 其他两个byte属性：通道ID  采集者
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列族名： " + familyName + ", 列名： " + columnName + ", 列值： " + value);
            }

        }
        table.close();
    }

    private Connection getConnection() throws IOException {
        //连接Hbase集群
        Configuration configuration = HBaseConfiguration.create();
//        configuration.set("hbase.zookeeper.quorum", "node01:2181,node02:2181,node03:2181");
        configuration.set("hbase.zookeeper.quorum", "10.1.52.3:2181,10.1.52.4:2181");

        Connection connection = ConnectionFactory.createConnection(configuration);

        return connection;
    }
}